package io.kimmking.common.rocketmq.config;

import com.google.common.collect.Lists;
import io.kimmking.common.rocketmq.base.BaseIConsumer;
import io.kimmking.common.rocketmq.base.RocketMQMsgListener;
import io.kimmking.common.rocketmq.config.consumer.StartConsumer;
import io.kimmking.dubbo.common.unit.ThrowableActionV2;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * 用于在服务启动时，扫描到的被添加@RocketMQMsgListener注解的bean，并启动消费者
 */
@Slf4j
@Component
public class MqMsgLoadListener implements ApplicationListener<ContextRefreshedEvent>, DisposableBean {

    private final List<StartConsumer> startConsumers = Lists.newArrayList();

    @Override
    public void destroy(){
        ThrowableActionV2.execute(() -> startConsumers.forEach(StartConsumer::close));
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        if(event.getApplicationContext().getParent() != null) {
            return;
        }
        //扫描到的被添加@RocketMQMsgListener注解的bean
        Map<String, Object> beans = event.getApplicationContext().getBeansWithAnnotation(RocketMQMsgListener.class);
        if(beans.isEmpty())
            return;
        //开始根据获取到的bean发起consumer操作
        beans.values().forEach(this::startConsumer);
    }

    private void startConsumer(Object bean) {
        if (!(bean instanceof BaseIConsumer)) {
            log.error("{} should be extends io.kimmking.common.rocketmq.base.BaseIConsumer", bean.getClass().getName());
            return;
        }
        //开始进行启动消费操作,此处需要获取bean<instanceof MessageListenerConcurrently>里的成员变量topic,consumer_group
        Field[] fs = bean.getClass().getDeclaredFields();
        ThrowableActionV2.execute(() -> {
            Field tf = Arrays.stream(fs).filter(field -> "topic".equals(field.getName())).findFirst().get();
            Field gf = Arrays.stream(fs).filter(field -> "consumer_group".equals(field.getName())).findFirst().get();
            tf.setAccessible(true);
            gf.setAccessible(true);
            startConsumers.add(new StartConsumer((MessageListenerConcurrently)bean,
                    String.valueOf(tf.get(bean)),
                    String.valueOf(gf.get(bean))));
        });

    }
}
